package com.lili.collection.core.runner.group;

import com.lili.collection.core.config.ProcessConfig;
import com.lili.collection.core.context.PipeContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.event.StageEvent;
import com.lili.collection.core.event.StageEventListener;
import com.lili.collection.core.event.StageEventListenerRegister;
import com.lili.collection.core.plugins.ProcessPlugin;
import com.lili.collection.core.runner.StageGroupRunner;
import lombok.Getter;

import java.util.ArrayList;
import java.util.List;

/**
 * 通用输入Runner
 *
 * @author lili
 * @date 2023/3/7 11:18
 */
public class GenericProcessorGroupRunner implements StageGroupRunner, StageEventListenerRegister {

    protected List<StageEventListener> stageContextChangedEventListener;

    @Getter
    private PipeContext pipeContext;

    private List<ProcessPlugin> processPluginList;

    public GenericProcessorGroupRunner(PipeContext pipeContext) {
        this.pipeContext = pipeContext;
    }

    @Override
    public void pause() {

    }

    @Override
    public void resume() {

    }

    @Override
    public void stop() {

    }

    @Override
    public void dispose() {

    }

    private void ensurePlugin() {
        if (this.processPluginList != null) return;
        this.processPluginList=new ArrayList<>();
        List<ProcessConfig> processConfigList = this.pipeContext.getContext().getProcessConfigList();
        for (ProcessConfig processConfig : processConfigList) {
            ProcessPlugin plugin = processConfig.getPlugin();
            plugin.init(processConfig);
            plugin.setPipeContext(this.pipeContext);
            this.processPluginList.add(plugin);
        }
    }

    @Override
    public void run() {
        this.ensurePlugin();
        StageContext stageContext = this.getPipeContext().getExchangeStageContext().take(StageType.Input);

        //处理数据
        for (ProcessPlugin processPlugin : processPluginList) {
            processPlugin.execute(stageContext);
        }

        this.getPipeContext().getExchangeStageContext().put(StageType.ProcessGroup, stageContext);
        triggerStageEvent(StageEvent.STAGE_CONTEXT_CHANGE, this.getStageType(), stageContext);
    }

    @Override
    public StageType getStageType() {
        return StageType.ProcessGroup;
    }

    @Override
    public synchronized void registerStageEventListener(StageEventListener listener) {
        ensureStageEventListener();
        this.stageContextChangedEventListener.add(listener);
    }

    @Override
    public void triggerStageEvent(StageEvent stageEvent, StageType stageType, Object params) {
        this.stageContextChangedEventListener.forEach(r -> r.onStageEventTrigger(stageEvent, this.getStageType(), params));
    }

    private void ensureStageEventListener() {
        if (this.stageContextChangedEventListener == null)
            this.stageContextChangedEventListener = new ArrayList<>();
    }
}
